package com.zhangdi.mgsync.core.rabbit;

import com.zhangdi.mgsync.core.message.EmailUtil;
import org.bson.BsonDocument;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.core.QueueInformation;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

/**
 * @author zhangdi
 * @version 1.0
 * @date 2021.11.6 15:11
 */
public class AdminTemplate {
    private final RabbitTemplate rabbitTemplate;
    private final AmqpAdmin amqpAdmin;
    private final EmailUtil emailUtil;
    private Exchange exchange;
    private Map<String, Queue> queues = new HashMap<>();
    private volatile Map<String, BsonDocument> queueResumeTokens = new HashMap<>();

    public EmailUtil getEmailUtil() {
        return emailUtil;
    }

    public QueueInformation getQueueInfo(String queueName){
        return amqpAdmin.getQueueInfo(queueName);
    }

    public void saveResumeToken(String key, BsonDocument token){
        queueResumeTokens.put(key, token);
    }

    public BsonDocument getResumeToken(String key){
        return queueResumeTokens.get(key);
    }

    public AdminTemplate(RabbitTemplate rabbitTemplate, AmqpAdmin amqpAdmin, EmailUtil emailUtil) {
        this.rabbitTemplate = rabbitTemplate;
        this.amqpAdmin = amqpAdmin;
        this.emailUtil = emailUtil;
    }

    public boolean create(String exchangeName, Collection<String> queues) throws AmqpException {
        amqpAdmin.declareExchange(createDirectExchange(exchangeName + ".direct"));
        for (String queueName : queues) {
            amqpAdmin.declareQueue(createQueue(queueName));
            amqpAdmin.declareBinding(BindingBuilder.bind(createQueue(queueName)).to(exchange).with(queueName).noargs());
        }
        return true;
    }

    public boolean create(String exchangeName, String queue) throws AmqpException {
        amqpAdmin.declareExchange(createDirectExchange(exchangeName + ".direct"));
        amqpAdmin.declareQueue(createQueue(queue));
        amqpAdmin.declareBinding(BindingBuilder.bind(createQueue(queue)).to(exchange).with(queue).noargs());
        return true;
    }

    public <T> void send(String collName, T data) {
        rabbitTemplate.convertAndSend(this.exchange.getName(), queues.get(collName).getName(), data);
    }

    public void send(String collName, Message message) {
        rabbitTemplate.send(this.exchange.getName(), queues.get(collName).getName(), message);
    }

    private Queue createQueue(String name) {
        Queue queue = QueueBuilder.durable(name).build();
        queues.put(name, queue);
        return queue;
    }

    private Exchange createDirectExchange(String name) {
        if (exchange == null) {
            return exchange = ExchangeBuilder.directExchange(name).durable(true).build();
        }
        return exchange;
    }
}
